package com.bzframework.pro.socket.manager;

import com.bzframework.pro.socket.MessagingRequest;
import com.bzframework.pro.socket.message.Message;
import com.bzframework.pro.socket.provider.SubscriptionProvider;
import org.springframework.util.AntPathMatcher;
import org.springframework.util.PathMatcher;
import reactor.core.publisher.Flux;


public class DefaultMessagingManager implements MessagingManager {

    private final static PathMatcher matcher = new AntPathMatcher();

    @Override
    public Flux<Message> subscribe(MessagingRequest request) {

        return Flux.defer(() -> {
            for (SubscriptionProvider provider :  SubscriptionProvider.providers.values()) {
                for (String pattern : provider.getTopicPattern()) {
                    if (matcher.match(pattern, request.getTopic())) {
                        return provider
                            .subscribe(request)
                            .map(v -> Message.success(request.getId(), request.getTopic(), v));
                    }
                }
            }
            return Flux.error(new RuntimeException("unsupported_topic"));
        });
    }

    public void register(SubscriptionProvider provider) {
        SubscriptionProvider.providers.put(provider.id(),provider);
    }

}
